package com.dw.etl.core.service;

import com.alibaba.fastjson.JSONObject;
import com.dw.etl.common.base.BaseException;
import com.dw.etl.common.base.BaseResult;
import com.dw.etl.common.constant.CommonResultConstant;
import com.dw.etl.common.util.LocalDateUtil;
import com.dw.etl.common.util.SnowflakeIdWorker;
import com.dw.etl.common.util.SpringBeanUtil;
import com.dw.etl.common.util.UUIDUtil;
import com.dw.etl.common.vo.FieldAttrVO;
import com.dw.etl.core.api.*;
import com.dw.etl.core.bean.model.*;
import com.dw.etl.core.bean.vo.DmSqlVO;
import com.dw.etl.core.bean.vo.JobQryVO;
import com.dw.etl.core.config.EtlConstant;
import com.dw.etl.core.mapper.MdMedTbFldMapper;
import com.dw.etl.core.utils.EtlUtil;
import io.swagger.annotations.ApiOperation;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.*;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;

import javax.annotation.Resource;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * Created by liujunlei on 2021/6/2.
 */
@Service
public class EtlRunServiceImpl implements EtlRunService {

    private static final Logger logger = LoggerFactory.getLogger(EtlRunServiceImpl.class);

    public static Map<String,Boolean> jobRunFlagMap = new HashMap<>();

    @Autowired
    BaseDbService baseDbService;
    @Autowired
    BaseDbTpcompService baseDbTpcompService;
    @Autowired
    EtlBaseFieldService etlBaseFieldService;
    @Autowired
    EtlBaseService etlBaseService;
    @Autowired
    TableOperService tableOperService;
    @Autowired
    private MdMedTbFldMapper mdMedTbFldMapper;
    @Autowired
    CommonService commonService;
    @Autowired
    EtlLogXxltobatchService etlLogXxltobatchService;
    @Autowired
    EtlIndexsqlService etlIndexsqlService;

    @Resource(name = "myJobLauncher")
    SimpleJobLauncher myJobLauncher;

    @Value("${dw.datasource.id}")
    private String dwDataSource;
    @Value("${dm.datasource.id}")
    private String dmDataSource;


    @Override
    public void autoCreatTable(String idBase) {
        EtlBase etlBase = etlBaseService.getById(idBase);
        String tableName = etlBase.getTbTar();
        EtlBaseField etlBaseField = new EtlBaseField();
        etlBaseField.setIdBase(idBase);
        List<EtlBaseField> etlBaseFieldList = etlBaseFieldService.findByModel(etlBaseField);
        BaseDb DbTar = baseDbService.getById(etlBase.getIdDbTar());
        BaseDbTpcomp dbTpcomp = new BaseDbTpcomp();
        dbTpcomp.setEuDbtp(DbTar.getEuTp());
        List<BaseDbTpcomp> dbTpcompList = baseDbTpcompService.findByModel(dbTpcomp);
        List<FieldAttrVO> list = tableOperService.findFieldMsgFromDb(DbTar, tableName);
        if (list.size() > 0) { // 存在执行更新

        } else {  //自动建表
            StringBuffer creatSql = new StringBuffer("create table " + tableName + "(");
            Map<String, BaseDbTpcomp> dbTpcompMap = new HashMap<>();
            boolean haveField = false;
            for (EtlBaseField field : etlBaseFieldList) {
                for (BaseDbTpcomp comp : dbTpcompList) {
                    if (field.getEuType().equals(comp.getEuJavatp())) {
                        if (field.getNumPrec() > 0 && comp.getFgDeci() > 0) { //小数
                            if (field.getEuType().equals("DATE")) {
                                creatSql.append(field.getField() + " " + comp.getEuFldtp() + ",");
                                haveField = true;
                                break;
                            } else if (field.getNumLength() < comp.getNumLength()) {
                                creatSql.append(field.getField() + " " + comp.getEuFldtp() + "(" + field.getNumLength() + "," + field.getNumPrec() + "),");
                                haveField = true;
                                break;
                            }

                        } else if (field.getNumPrec() == 0 && comp.getFgDeci() == 0) {
                            if (field.getEuType().equals("DATE")) {
                                creatSql.append(field.getField() + " " + comp.getEuFldtp() + ",");
                                haveField = true;
                                break;
                            } else if (field.getNumLength() < comp.getNumLength()) {
                                creatSql.append(field.getField() + " " + comp.getEuFldtp() + "(" + field.getNumLength() + "),");
                                haveField = true;
                                break;
                            }

                        }
                    }
                }
            }
            if (haveField) { //可以创建
                creatSql = creatSql.deleteCharAt(creatSql.length() - 1);
                creatSql.append(")");
                tableOperService.execDdlSqlByDb(creatSql.toString(), etlBase.getIdDbTar().toString());
            }
        }
    }

    @Override
    public void createTable(String idTb) {
        MdMedTbFld mdMedTbFld = new MdMedTbFld();
        mdMedTbFld.setIdTb(idTb);
        List<MdMedTbFld> fldList = mdMedTbFldMapper.findByModel(mdMedTbFld);
        StringBuffer buffer = new StringBuffer();
        if (fldList.size() > 0) {
            BaseDbTpcomp dbTpcomp = new BaseDbTpcomp();
            dbTpcomp.setEuDbtp("mysql");
            List<BaseDbTpcomp> dbTpcompList = baseDbTpcompService.findByModel(dbTpcomp);
            for (MdMedTbFld fld : fldList) {
                if (buffer.length() == 0) {
                    buffer.append("create table " + fld.getIdTb() + "(");
                }
                if ("pk".equals(fld.getIdPubfld())) {
                    buffer.append(fld.getIdPubfld() + " bigint(30) primary key,");
                } else {
                    buffer.append(fld.getIdPubfld() + " " + EtlUtil.getDbTypeByStandType(dbTpcompList, fld.getEuTp(), fld.getNumLth(), fld.getNumPrec()) + ",");
                }
            }
            buffer.append("dt_sys_cre date,dt_sys_modi date)");
            tableOperService.execDdlSqlByDb(buffer.toString(),dmDataSource);
        }
    }

    @Override
    public void runEtlBaseJobByDay(String nowDate,String sdJbcls,String xxlJObName,Date triggerTime) {
        JobParametersBuilder parametersBuilder = new JobParametersBuilder();
        commonService.addOrgJobParameter(parametersBuilder);
        EtlUtil.addDayParameter(nowDate,parametersBuilder);
        EtlBase etlBaseModel = new EtlBase();
        etlBaseModel.setNaJbcls(sdJbcls);
        etlBaseModel.setStatus((byte)1);
        List<EtlBase> etlBaseList = etlBaseService.findAllByModel(etlBaseModel);
        for(EtlBase etlBase:etlBaseList) {
            runJob(etlBase.getIdBase(),etlBase.getIdDbSou()+"",etlBase.getIdDbTar()+"",etlBase.getQsql(),etlBase.getDel(),etlBase.getTbTar(), etlBase.getIdJb(),triggerTime,parametersBuilder);
        }
    }

    @Override
    public void runDmJObByDay(String nowDate,String xxlJObName,Date triggerTime) {
        JobParametersBuilder parametersBuilder = new JobParametersBuilder();
        commonService.addOrgJobParameter(parametersBuilder);
        EtlUtil.addDayParameter(nowDate,parametersBuilder);
        List<DmSqlVO> sqlList = etlIndexsqlService.findAllSql();
        for(DmSqlVO sqlVO:sqlList) {
            runJob(sqlVO.getIdIndex(), dwDataSource, dmDataSource, sqlVO.getQrySql(), sqlVO.getDelSql(), sqlVO.getTarTableName(), EtlConstant.NA_BATCH_JOB_DM, triggerTime, parametersBuilder);
        }
    }

    @Autowired
    EtlLogHandbatchService etlLogHandbatchService;
    @Autowired
    EtlLogHandbatchItemService etlLogHandbatchItemService;

    @Override
    public void runHistoryJob(List<EtlBase> etlBaseList,List<DmSqlVO> dmSqlVOList,JobQryVO jobQryVO,JobParametersBuilder parametersBuilder){
        int counterStop = 0;//计算器，防止死循环，无实际意义
        try{
            String dBegin = jobQryVO.getdBegin();
            String dEnd = jobQryVO.getdEnd();
            if(StringUtils.isEmpty(dBegin) || StringUtils.isEmpty(dBegin)) {
                throw new BaseException("时间不能为空");
            }
            Date dtBegin = LocalDateUtil.dateFormat.parse(dBegin+" 00:00:00");
            Date dtEnd = LocalDateUtil.dateFormat.parse(dEnd+" 00:00:00");
            Date dtNext;
            List<EtlLogHandbatchItem> handbatchItemList;
            EtlLogHandbatchItem handbatchItem;
            EtlLogHandbatch etlLogHandbatch = new EtlLogHandbatch();
            etlLogHandbatch.setDtBegin(dtBegin);
            etlLogHandbatch.setDtEnd(dtEnd);
            etlLogHandbatch.setNa(jobQryVO.getNa());
            if(jobQryVO.getIdBaseList()!=null) {
                etlLogHandbatch.setParam(JSONObject.toJSONString(jobQryVO.getIdBaseList()));
            } else if(jobQryVO.getIdIndexList()!=null) {
                etlLogHandbatch.setParam(JSONObject.toJSONString(jobQryVO.getIdIndexList()));
            }
            Date functionBeginDate = new Date();
            Long days = (dtEnd.getTime()-dtBegin.getTime())/1000;
            etlLogHandbatch.setNumDay(Integer.parseInt(days.toString()));
            JobExecution jobExecution;
            while(dtBegin.before(dtEnd) && counterStop<10000) {
                if(jobRunFlagMap.get(jobQryVO.getNa())) {
                    break;
                }
                handbatchItemList = new ArrayList<>();
                dtNext = EtlUtil.getNextDate(dtBegin,3);
                EtlUtil.addDayParameter(LocalDateUtil.dayFormat.format(dtNext),parametersBuilder);
                counterStop++;
                etlLogHandbatch.setNumDayFinish(counterStop);
                if(counterStop==1) {
                    etlLogHandbatch.setIdHandbatch(UUIDUtil.getUUID32());
                    etlLogHandbatchService.insert(etlLogHandbatch);
                } else {
                    etlLogHandbatchService.updateByIdSelective(etlLogHandbatch);
                }
                if(etlBaseList!=null &&etlBaseList.size()>0) {
                    for(EtlBase etlBase:etlBaseList) {
                        jobExecution = runJob(etlBase.getIdBase(),etlBase.getIdDbSou()+"",etlBase.getIdDbTar()+"",etlBase.getQsql(),etlBase.getDel(),etlBase.getTbTar(), etlBase.getIdJb(),null,parametersBuilder);
                        if(jobExecution!=null) {
                            handbatchItem = new EtlLogHandbatchItem();
                            handbatchItem.setIdHendbatch(etlLogHandbatch.getIdHandbatch());
                            handbatchItem.setIdJobExecution(jobExecution.getId());
                            handbatchItem.setNaBatchjob(jobExecution.getJobInstance().getJobName());
                            handbatchItem.setStatus(jobExecution.getStatus().name());
                            handbatchItemList.add(handbatchItem);
                        }
                    }
                }
                if(dmSqlVOList!=null &&dmSqlVOList.size()>0) {
                    for(DmSqlVO sqlVO:dmSqlVOList) {
                        jobExecution = runJob(sqlVO.getIdIndex(), dwDataSource, dmDataSource, sqlVO.getQrySql(), sqlVO.getDelSql(), sqlVO.getTarTableName(), EtlConstant.NA_BATCH_JOB_DM, null, parametersBuilder);
                        if(jobExecution!=null) {
                            handbatchItem = new EtlLogHandbatchItem();
                            handbatchItem.setIdHendbatch(etlLogHandbatch.getIdHandbatch());
                            handbatchItem.setIdJobExecution(jobExecution.getId());
                            handbatchItem.setNaBatchjob(jobExecution.getJobInstance().getJobName());
                            handbatchItem.setStatus(jobExecution.getStatus().name());
                            handbatchItemList.add(handbatchItem);
                        }
                    }
                }
                dtBegin = dtNext;
                if(handbatchItemList.size()>0) {
                    etlLogHandbatchItemService.insertList(handbatchItemList);
                }
            }
            Date functionEndDate = new Date();
            Long timesDiff = functionEndDate.getTime()-functionBeginDate.getTime();
            etlLogHandbatch.setTimes(timesDiff);
            etlLogHandbatch.setDesTimes(LocalDateUtil.dateDiffString(functionBeginDate,functionEndDate));
            etlLogHandbatchService.updateByIdSelective(etlLogHandbatch);
        } catch (Exception e) {
            logger.error(e.getStackTrace().toString());
        } finally {
            jobRunFlagMap.remove(jobQryVO.getNa());
        }
    }

    private JobExecution runJob(String sourceId,String idDbSou,String idDbTar,String qSql,String delSql,String tbTar,String jobName,Date triggerTime,JobParametersBuilder parametersBuilder) {
        JobParameters parameters = parametersBuilder.addString("tableName",tbTar)
                .addString("sourceId",sourceId)
                .addString("idDbSou",idDbSou)
                .addString("idDbTar",idDbTar)
                .addLong("UUID", SnowflakeIdWorker.getId())
                .addString("qSql",paramReplace(parametersBuilder,qSql)).toJobParameters();
        if(StringUtils.isNotEmpty(delSql)) {
            tableOperService.execDelByDb(paramReplace(parametersBuilder,delSql),tbTar+"");
        }
        try {
            Job job = (Job) SpringBeanUtil.getBean(jobName);
            JobExecution jobExecution = myJobLauncher.run(job,parameters);
            if(triggerTime!=null) {
                EtlLogXxltobatch xxltobatch = new EtlLogXxltobatch();
                xxltobatch.setNaHandler(jobName);
                xxltobatch.setTriggerTime(triggerTime);
                xxltobatch.setJobInstanceId(jobExecution.getJobInstance().getInstanceId());
                xxltobatch.setJobName(jobExecution.getJobInstance().getJobName());
                xxltobatch.setStatus(jobExecution.getStatus().name());
                xxltobatch.setDtSysCre(new Date());
                if(BatchStatus.COMPLETED.name().equals(xxltobatch.getStatus())) {
                    xxltobatch.setFgHandle(true);
                } else {
                    xxltobatch.setFgHandle(false);
                }
                etlLogXxltobatchService.insert(xxltobatch);
            }
            return jobExecution;
        } catch (Exception e) {
            logger.error(e.getStackTrace().toString());
        }
        return null;
    }

    private String paramReplace(JobParametersBuilder parametersBuilder ,String sql) {
        Map<String,JobParameter> parameterMap = parametersBuilder.toJobParameters().getParameters();
        JobParameter parameter;
        for(String key:parameterMap.keySet()) {
            parameter = parameterMap.get(key);
            if(sql.contains("{"+key+"}")) {
                if(key.startsWith("DT_")) {
                    sql = sql.replaceAll("'\\{"+key+"\\}'","to_date('"+parameter.getValue().toString()+"','yyyymmdd hh24:mi:ss')");
                } else {
                    sql = sql.replaceAll("\\{"+key+"\\}",parameter.getValue().toString());
                }

            }
        }
        return sql;
    }
}
